package com.zdxlz.fcmp.common.mqtt.handler;

import jakarta.annotation.Resource;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.task.TaskExecutor;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author: aman
 * @date 2025/2/14 11:32
 * @description
 */

@Slf4j
@Component
public class MqttMessageRouter {

    @Resource
    private TaskExecutor mqttMsgTaskExecutor;

    private MqttMessageHandler defaultHandler;
    private final List<RouteRule> routeRules = new CopyOnWriteArrayList<>();
    private final Map<String, MqttMessageHandler> handlerMap = new ConcurrentHashMap<>();

    private final Map<String, CompletableFuture<?>> futureRegistry = new ConcurrentHashMap<>();


    // 设置默认处理器（处理未注册的 Topic）
    public void setDefaultHandler(MqttMessageHandler handler) {
        this.defaultHandler = handler;
    }

    public void registerWildcardHandler(String topicFilter, MqttMessageHandler handler) {
        routeRules.add(new RouteRule(topicFilter, handler));
    }

    public void routeMessage(String incomingTopic, byte[] payload) {
        routeRules.stream()
                .filter(rule -> matches(rule.topicFilter, incomingTopic))
                .findFirst()
                .ifPresentOrElse(
                        rule -> rule.handler.handleMessage(incomingTopic, payload),
                        () -> defaultHandler.handleMessage(incomingTopic, payload)
                );
    }


    // 提交任务到线程池并注册 Future
    public void routeAsync(String topic, byte[] payload) {

        routeRules.stream()
                .filter(rule -> matches(rule.topicFilter, topic))
                .findFirst()
                .ifPresentOrElse(rule -> CompletableFuture
                        .runAsync(() -> rule.handler.handleMessage(topic, payload), mqttMsgTaskExecutor)
                        .whenComplete((result, ex) -> {
                            if (ex != null) {
                                log.error("处理失败【{}】: {}", topic, ex.getMessage());
                            }
                            futureRegistry.remove(topic); // 清理完成的任务
                        }), () -> defaultHandler.handleMessage(topic, payload));

    }

    // 根据 Topic 获取 Future（供外部查询）
    public Optional<CompletableFuture<?>> getFuture(String topic) {
        return Optional.ofNullable(futureRegistry.get(topic));
    }


    // MQTT 通配符匹配逻辑（# = 多级，+ = 单级）
    private boolean matches(String filter, String topic) {
        String regex = filter
                .replaceAll("\\+", "[^/]+")
                .replaceAll("/#", "(/.*)?");
        regex = "^" + regex + "$";
        return topic.matches(regex);
    }

    @Data
    @AllArgsConstructor
    private static class RouteRule {
        String topicFilter;
        MqttMessageHandler handler;
    }
}
